﻿// Copyright (c) Inossem. All rights reserved.
// Licensed under the LGPL-3.0 license. See LICENSE file in the project root for full license information.
using System;
using System.Collections.Generic;
using System.IO.Pipes;
using System.Linq;
using System.Runtime.Serialization;
using System.Text;
using System.Threading;
using NamedPipeWrapper.IO;
using NamedPipeWrapper.Threading;
using System.Collections.Concurrent;
using NamedPipeWrapper.Common;
using System.Threading.Tasks;

namespace NamedPipeWrapper
{
    /// <summary>
    /// Represents a connection between a named pipe client and server.
    /// </summary>
    /// <typeparam name="TRead">Reference type to read from the named pipe</typeparam>
    /// <typeparam name="TWrite">Reference type to write to the named pipe</typeparam>
    public class NamedPipeConnectionAsync<TRead, TWrite>
        where TRead : class
        where TWrite : class
    {

        public readonly int Id;

        /// <summary>
        /// Gets the connection's name.
        /// </summary>
        public string Name { get; }

        /// <summary>
        /// Gets a value indicating whether the pipe is connected or not.
        /// </summary>
        public bool IsConnected { get { return _streamWrapper.IsConnected; } }

        /// <summary>
        /// Invoked when the named pipe connection terminates.
        /// </summary>
        public event ConnectionAsyncEventHandler<TRead, TWrite> Disconnected;

        /// <summary>
        /// Invoked whenever a message is received from the other end of the pipe.
        /// </summary>
        public event ConnectionAsyncMessageEventHandler<TRead, TWrite> ReceiveMessage;

        /// <summary>
        /// Invoked when an exception is thrown during any read/write operation over the named pipe.
        /// </summary>
        public event ConnectionAsyncExceptionEventHandler<TRead, TWrite> Error;

        private PipeStreamWrapper<TRead, TWrite> _streamWrapper;

        private bool _notifiedSucceeded;

        internal NamedPipeConnectionAsync(int id, string name, PipeStream serverStream)
        {
            Id = id;
            Name = name;
            _streamWrapper = new PipeStreamWrapper<TRead, TWrite>(serverStream);
        }

        protected PipeStreamWrapper<TRead, TWrite> StreamWrapper
        {
            get
            {
                return this._streamWrapper;
            }
        }

        public virtual void Close()
        {
            closePipe();
            OnClose();
        }

        /// <summary>
        ///     Invoked on the background thread.
        /// </summary>
        private void closePipe()
        {
            if (_streamWrapper != null)
            {
                _streamWrapper.Close();
            }
            else
            {
                throw new Exception("Stream is Closed.");
            }
        }

        /// <summary>
        ///     Invoked on the UI thread.
        /// </summary>
        protected void OnClose()
        {
            // Only notify observers once
            if (_notifiedSucceeded)
                return;

            _notifiedSucceeded = true;

            if (Disconnected != null)
                Disconnected(this);
        }

        /// <summary>
        ///     Invoked on the UI thread.
        /// </summary>
        /// <param name="exception"></param>
        protected void OnError(Exception exception)
        {
            if (Error != null)
                Error(this, exception);
        }

        protected void OnReceiveMessage(TRead message)
        {
            if (ReceiveMessage != null)
                ReceiveMessage(this, message);
        }

        /// <summary>
        /// Begins reading from and writing to the named pipe on a background thread.
        /// This method returns immediately.
        /// </summary>
        public virtual void Open()
        {
            Task.Run(readPipeAsync);
        }

        public void PushMessage(TWrite message)
        {
            if (message == null)
            {
                LogHelper.Info("PushMessage is null");
                return;
            }

            writePipe(message);
        }

        /// <summary>
        /// Adds the specified <paramref name="message"/> to the write queue.
        /// The message will be written to the named pipe by the background thread
        /// at the next available opportunity.
        /// </summary>
        /// <param name="message"></param>
        public async Task PushMessageAsync(TWrite message)
        {
            if (message == null)
            {
                LogHelper.Info("PushMessage is null");
                return;
            }

            await writePipeAsync(message);
        }

        private async Task readPipeAsync()
        {
            while (IsConnected && StreamWrapper.CanRead)
            {
                try
                {
                    var obj = await StreamWrapper.ReadObjectAsync();
                    if (obj != null)
                    {
                        this.OnReceiveMessage(obj);
                    }
                }
                catch (Exception ex)
                {
                    LogHelper.Error(ex);
                }
            }

            closePipe();
            OnClose();
        }

        private void writePipe(TWrite message)
        {
            if (IsConnected && StreamWrapper.CanWrite)
            {
                try
                {
                    if (message != null)
                    {
                        StreamWrapper.WriteObject(message);
                        StreamWrapper.WaitForPipeDrain();
                    }
                }
                catch (Exception ex)
                {
                    LogHelper.Error(ex);
                    OnError(ex);
                }
            }
            else
            {
                closePipe();
                OnClose();
            }
        }

        /// <summary>
        ///     Invoked on the background thread.
        /// </summary>
        /// <exception cref="SerializationException">An object in the graph of type parameter <typeparamref name="TWrite"/> is not marked as serializable.</exception>
        private async Task writePipeAsync(TWrite message)
        {
            if (IsConnected && StreamWrapper.CanWrite)
            {
                try
                {
                    if (message != null)
                    {
                        await StreamWrapper.WriteObjectAsync(message);
                        StreamWrapper.WaitForPipeDrain();
                    }
                }
                catch (Exception ex)
                {
                    LogHelper.Error(ex);
                    OnError(ex);
                }
            }
            else
            {
                closePipe();
                OnClose();
            }
        }
    }

    static class ConnectionAsyncFactory
    {
        private static int _lastId;

        public static NamedPipeConnectionAsync<TRead, TWrite> CreateConnection<TRead, TWrite>(PipeStream pipeStream)
            where TRead : class
            where TWrite : class
        {
            return new NamedPipeConnectionAsync<TRead, TWrite>(++_lastId, "ClientAsync " + _lastId, pipeStream);
        }
    }

    /// <summary>
    /// Handles new connections.
    /// </summary>
    /// <param name="connection">The newly established connection</param>
    /// <typeparam name="TRead">Reference type</typeparam>
    /// <typeparam name="TWrite">Reference type</typeparam>
    public delegate void ConnectionAsyncEventHandler<TRead, TWrite>(NamedPipeConnectionAsync<TRead, TWrite> connection)
        where TRead : class
        where TWrite : class;

    /// <summary>
    /// Handles messages received from a named pipe.
    /// </summary>
    /// <typeparam name="TRead">Reference type</typeparam>
    /// <typeparam name="TWrite">Reference type</typeparam>
    /// <param name="connection">Connection that received the message</param>
    /// <param name="message">Message sent by the other end of the pipe</param>
    public delegate void ConnectionAsyncMessageEventHandler<TRead, TWrite>(NamedPipeConnectionAsync<TRead, TWrite> connection, TRead message)
        where TRead : class
        where TWrite : class;

    /// <summary>
    /// Handles exceptions thrown during read/write operations.
    /// </summary>
    /// <typeparam name="TRead">Reference type</typeparam>
    /// <typeparam name="TWrite">Reference type</typeparam>
    /// <param name="connection">Connection that threw the exception</param>
    /// <param name="exception">The exception that was thrown</param>
    public delegate void ConnectionAsyncExceptionEventHandler<TRead, TWrite>(NamedPipeConnectionAsync<TRead, TWrite> connection, Exception exception)
        where TRead : class
        where TWrite : class;
}
